package com.isyscore.os.metadata.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.ImmutableMap;
import com.isyscore.boot.login.LoginUserManagerImpl;
import com.isyscore.boot.mybatis.PageRequest;
import com.isyscore.os.core.entity.SqlQuery;
import com.isyscore.os.core.exception.DataFactoryException;
import com.isyscore.os.core.exception.ErrorCode;
import com.isyscore.os.core.util.InitiallyUtils;
import com.isyscore.os.metadata.constant.CommonConstant;
import com.isyscore.os.metadata.constant.DataSourceConstant;
import com.isyscore.os.metadata.dao.DataSourceMapper;
import com.isyscore.os.metadata.database.AbstractDatabase;
import com.isyscore.os.metadata.database.KingbaseDatabase;
import com.isyscore.os.metadata.database.PGSQLDatabase;
import com.isyscore.os.metadata.database.SqlServerDatabase;
import com.isyscore.os.metadata.manager.DataSourceManager;
import com.isyscore.os.metadata.manager.DatabaseManager;
import com.isyscore.os.metadata.model.dto.*;
import com.isyscore.os.metadata.model.entity.DataSource;
import com.isyscore.os.metadata.model.query.DataSourceQuery;
import com.isyscore.os.metadata.model.query.DataSourceQueryWrapper;
import com.isyscore.os.metadata.model.query.IDataSourceQuery;
import com.isyscore.os.metadata.model.vo.*;
import com.isyscore.os.metadata.service.DataFlowDefinitionService;
import com.isyscore.os.metadata.service.DataSourceService;
import com.isyscore.os.metadata.service.SqlQueryService;
import com.isyscore.os.metadata.service.TaskService;
import com.isyscore.os.metadata.utils.AesUtil;
import com.isyscore.os.metadata.utils.UDMPFileUtils;
import com.isyscore.os.metadata.utils.UdmpBeanUtils;
import com.isyscore.os.permission.common.constants.PermissionConstants;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.web.multipart.MultipartFile;

import javax.validation.constraints.NotNull;
import java.util.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

import static com.isyscore.os.core.exception.ErrorCode.DATA_SOURCE_PASS_ENCRYPT_ERROR;
import static com.isyscore.os.metadata.enums.DataSourceTypeEnum.*;
import static java.util.Collections.emptyList;

@Slf4j
@Service
public class DataSourceServiceImpl extends ServiceImpl<DataSourceMapper, DataSource> implements DataSourceService {

    @Autowired
    private DataSourceMapper dataSourceMapper;
    @Autowired
    private DataSourceService dataSourceService;
    @Autowired
    private TaskService taskService;

    @Autowired
    private DatabaseManager databaseManager;
    @Autowired
    private DataSourceManager dataSourceManager;

    @Autowired
    private LoginUserManagerImpl loginUserManager;
    @Autowired
    private DataFlowDefinitionService dataFlowDefinitionService;
    @Autowired
    private SqlQueryService sqlQueryService;
    @Override
    public void lock(String dataSourceId, String databaseName, Integer isLocked) {
        LambdaQueryWrapper<DataSource> query = Wrappers.<DataSource>lambdaQuery()
                .eq(DataSource::getId, dataSourceId)
                .and(wrapper -> wrapper
                        .eq(DataSource::getTenantId, loginUserManager.getCurrentTenantId())
                        .or()
                        .eq(DataSource::getTenantId, PermissionConstants.GLOBAL_BUSINESS_DATA_TENANT_ID_KEY));
        DataSource dataSource = InitiallyUtils.markInitially(() -> getOne(query));
        if (Objects.isNull(dataSource)) {
            throw new DataFactoryException(ErrorCode.UPDATE_NOT_FOUNT);
        }
        AtomicBoolean flag = new AtomicBoolean(false);
        List<DatabasePair> list = JSON.parseArray(dataSource.getDatabaseConfig(), DatabasePair.class);
        //兼容旧版本数据
        if (CollectionUtils.isEmpty(list)) {
            list = new ArrayList<>();
            // database_name -> database_config
            List<String> databaseNames = JSONArray.parseArray(dataSource.getDatabaseName(), String.class);
            list.addAll(databaseNames.parallelStream().map(name -> new DatabasePair(name, 1)).collect(Collectors.toList()));
        }
        List<DatabasePair> res = list.parallelStream().map(pair -> {
            if (Objects.equals(pair.getDatabaseName(), databaseName)) {
                flag.set(true);
                if (Objects.equals(pair.getIsLocked(), isLocked)) {
                    throw new DataFactoryException(ErrorCode.ALREADY_LOCKED);
                } else {
                    pair.setIsLocked(isLocked);
                }
                return pair;
            }
            return pair;
        }).collect(Collectors.toList());

        if (Objects.equals(flag.get(), false)) {
            throw new DataFactoryException(ErrorCode.UPDATE_NOT_FOUNT);
        }
        dataSource.setDatabaseConfig(JSON.toJSONString(res));
        InitiallyUtils.markInitially(() -> update(dataSource, query));
    }


    @Override
    @Transactional(rollbackFor = Exception.class)
    public void addDataSource(DataSourceDTO dataSourceDTO) {
        if (Objects.equals(dataSourceDTO.getType(), ORACLE.getCode())) {
            try {
                Assert.notNull(dataSourceDTO.getConnectType(), () -> "oracle连接类型不能为空");
                Assert.notNull(dataSourceDTO.getBasicType(), () -> "oracle连接方式不能为空");
                Assert.notNull(dataSourceDTO.getBasicValue(), () -> "oracle SID 或 ServerName 不能为空");
            } catch (IllegalArgumentException e) {
                throw new DataFactoryException(ErrorCode.PARAM_ERROR, e.getMessage());
            }
        }
        DataSource dataSource = new DataSource();
        if (!CollectionUtils.isEmpty(dataSourceDTO.getDatabaseNameList())) {
            dataSourceDTO.setDatabaseConfig(JSON.toJSONString(dataSourceDTO.getDatabaseNameList()
                    .parallelStream().map(name -> new DatabasePair(name, 1)).collect(Collectors.toList())));
            dataSource = UdmpBeanUtils.copy(dataSourceDTO, DataSource.class);
            dataSource.setDatabaseName(JSONObject.toJSONString(dataSourceDTO.getDatabaseNameList()));
        }
        try {
            dataSource.setPassword(AesUtil.encrypt(dataSource.getPassword()));
            save(dataSource);
        } catch (Exception e) {
            throw new DataFactoryException(DATA_SOURCE_PASS_ENCRYPT_ERROR);
        }


    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean updateDataSource(DataSourceDTO dataSourceDTO) {
        if (Objects.isNull(getById(dataSourceDTO.getId()))) {
            throw new DataFactoryException(ErrorCode.UPDATE_NOT_FOUNT);
        }
        DataSource dataSource = UdmpBeanUtils.copy(dataSourceDTO, DataSource.class);
        if (!BeanUtil.isEmpty(dataSourceDTO.getDatabaseNameList())) {
            dataSource.setDatabaseConfig(JSON.toJSONString(dataSourceDTO.getDatabaseNameList()
                    .parallelStream().map(name -> new DatabasePair(name, 1)).collect(Collectors.toList())));
            dataSource.setDatabaseName(JSONObject.toJSONString(dataSourceDTO.getDatabaseNameList()));
        }
        updateById(dataSource);
        return true;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean delBatch(List<Long> datasourceIds) {
        if (CollectionUtils.isEmpty(datasourceIds)) {
            return false;
        }
        List<String> kettleUsedDatasourceIds= dataFlowDefinitionService.getDataSourceIdFromFlow();
        for(Long datasourceId: datasourceIds){
            if(kettleUsedDatasourceIds.contains(String.valueOf(datasourceId))){
                throw new DataFactoryException(ErrorCode.DATABASE_REFERENCED_BY_KETTLE);
            }
        }
        List<Long> sqlQueryDataSourceIds= sqlQueryService.list().stream().map(SqlQuery::getDataSourceId).distinct().collect(Collectors.toList());
        for(Long datasourceId: datasourceIds){
            if(sqlQueryDataSourceIds.contains(datasourceId)){
                throw new DataFactoryException(ErrorCode.DATABASE_REFERENCED_BY_SQL_QUERY);
            }
        }

        for (Long datasourceId : datasourceIds) {
            dataSourceManager.removeDataSource(dataSourceService.getDataSource(datasourceId));
        }
        removeByIds(datasourceIds);
        return true;
    }

    //checked
    @Override
    public IPage<DataSourceDTO> getList(DataSourceQueryWrapper query) {
        if (query.getPageSize() == null) {
            query.setPageSize(CommonConstant.DEFAULT_PAGE_SIZE);
        }
        if (query.getPageNo() == null) {
            query.setPageNo(CommonConstant.DEFAULT_PAGE_NO);
        }
        QueryWrapper<DataSource> queryWrapper = new QueryWrapper();

        if (query.getType() != null) {
            //2021.06.22 针对前段可以多选type进行参数适配
            List<Integer> types = Arrays.stream(query.getType().split(","))
                    .map(Integer::valueOf).collect(Collectors.toList());
            queryWrapper.in("type", types);
        }
        if (StringUtils.isNotBlank(query.getName())) {
            queryWrapper.and(wrapper -> wrapper.like("name", query.getName()).or().like("database_name", query.getName()));
        }
        queryWrapper.orderByDesc("update_time");
        queryWrapper.and(wrapper ->
                wrapper.eq("tenant_id", loginUserManager.getCurrentTenantId())
                        .or()
                        .eq("tenant_id", PermissionConstants.GLOBAL_BUSINESS_DATA_TENANT_ID_KEY));
        IPage<DataSource> page = InitiallyUtils.markInitially(() -> dataSourceMapper.selectPage(new Page<>(query.getPageNo(), query.getPageSize()), queryWrapper));
        return page.convert(this::toDataSourceDTO);
    }

    @Override
    public List<String> test(DataSourceDTO dataSourceDTO) {
        AbstractDatabase db = DatabaseManager.findDb(dataSourceDTO);
        db.execSql(dataSourceDTO, "", null);
        return null;
//        BaseSource source = SourceFactory.getSource(name);
//        return source.test(dataSourceDTO);
    }

    @Override
    public List<DataSourceDTO> getAll() {
        QueryWrapper<DataSource> queryWrapper = new QueryWrapper<>();
        queryWrapper.and(wrapper ->
                wrapper.eq("tenant_id", loginUserManager.getCurrentTenantId())
                        .or()
                        .eq("tenant_id", PermissionConstants.GLOBAL_BUSINESS_DATA_TENANT_ID_KEY));
        queryWrapper.orderByDesc("update_time");
        return InitiallyUtils.markInitially(() -> list(queryWrapper)).parallelStream()
                .map(this::toDataSourceDTONoPage)
                .filter(Objects::nonNull)//过滤掉因连接出错而返回的空值
                .collect(Collectors.toList());
    }

    @Override
    public ResultVO getTableList(Long dataSourceId, String databaseName, boolean joinTenant) {
        DataSourceDTO dataSourceDTO = get(dataSourceId, joinTenant);
        dataSourceDTO.setSelectDatabaseName(databaseName);
        return databaseManager.getTableList(dataSourceDTO);

//        BaseSource source = SourceFactory.getSource(DataSourceTypeEnum.getType(dataSourceDTO.getType()).getName());
//        return source.getTableList(dataSourceDTO);
    }

    @Override
    public ResultVO getTableListWithAlias(Long dataSourceId, String databaseName, boolean joinTenant) {
        DataSourceDTO dataSourceDTO = get(dataSourceId, joinTenant);
        dataSourceDTO.setSelectDatabaseName(databaseName);
        return databaseManager.getTableListWithAlias(dataSourceDTO);

    }

    @Override
    public ResultVO getTableStruct(Long dataSourceId, String databaseName, String tableName) {
        DataSourceDTO dataSourceDTO = get(dataSourceId, true);
        dataSourceDTO.setSelectDatabaseName(databaseName);
        dataSourceDTO.setTableName(tableName);
        ResultVO resultVO = databaseManager.getTableStruct(dataSourceDTO);
        addStatisticsType(resultVO.getContent());
        return resultVO;
    }

    @Override
    public ResultVO getTableData(Long dataSourceId, IDataSourceQuery query) {
        DataSourceDTO dataSourceDTO = get(dataSourceId, true);
        DataQueryVO dataQueryVO = new DataQueryVO();
        if (query instanceof DataSourceQuery) {
            DataSourceQuery dataSourceQuery = (DataSourceQuery) query;
            dataSourceDTO.setSelectDatabaseName(dataSourceQuery.getDatabaseName());
            dataSourceDTO.setTableName(dataSourceQuery.getTableName());
            dataQueryVO.setCurrent(dataSourceQuery.getPageNo());
            dataQueryVO.setSize(dataSourceQuery.getPageSize());
        } else if (query instanceof DataQueryVO) {
            dataQueryVO = (DataQueryVO) query;
            dataSourceDTO.setSelectDatabaseName(dataQueryVO.getDatabaseName());
            dataSourceDTO.setTableName(dataQueryVO.getTableName());
        }
        return databaseManager.getTableData(dataSourceDTO, dataQueryVO);
    }

    //todo
    @Override
    public List<ResultVO> getLinkTableList(Long dataSourceId, String databaseName, String tableName, String taskId) {
        //-----------汤老师给出的联想算法和排序算法
        //联想关联表算法：根据中心表名+主键，或者带id(id除外)的字段去进行表名搜索
        //关联表排序逻辑：1.(表名+id)优先:强关联，2.(表名带dim)维表其次, 3.其他表排后面：弱关联
        //获取中心表结构
        try {
            log.info("租户ID为{}", loginUserManager.getCurrentLoginUser().getTenantId());
            DataSourceDTO dataSourceDTO = get(dataSourceId, true);
            if (Objects.isNull(dataSourceDTO)) {
                throw new DataFactoryException(ErrorCode.DATASOURCE_NOT_FOUND);
            }
            dataSourceDTO.setSelectDatabaseName(databaseName);
            dataSourceDTO.setTableName(tableName);

            ResultVO resultVO = databaseManager.getTableStruct(dataSourceDTO);
//            BaseSource source = SourceFactory.getSource(DataSourceTypeEnum.getType(dataSourceDTO.getType()).getName());
//            ResultVO resultVO = source.getTableStruct(dataSourceDTO);
            List<ResultVO> list = new ArrayList<>();
            List<ResultVO> strongLinkList = new ArrayList<>();
            List<ResultVO> weakLinkList = new ArrayList<>();
            List<ResultVO> dimLinkList = new ArrayList<>();
            resultVO.setTableName(tableName);
            resultVO.setHead(null);
            addStatisticsType(resultVO.getContent());
            list.add(resultVO);
            taskService.setTaskProgress(taskId, 20.00);
            List<Map<String, Object>> content = resultVO.getContent();
            List<String> columnList = new ArrayList<>();
            //查找中心表主键
            String primaryKey = null;
            for (Map<String, Object> map : content) {
                if (Objects.equals(dataSourceDTO.getType(), MYSQL.getCode())) {
                    if (Objects.equals(DataSourceConstant.MYSQL_KEY, map.get("columnKey"))) {
                        primaryKey = map.get("columnName").toString();
                    }
                } else if (Objects.equals(dataSourceDTO.getType(), CLICKHOUSE.getCode())) {
                    if (Objects.equals(DataSourceConstant.CLICKHOUSE_KEY, map.get("columnKey"))) {
                        primaryKey = map.get("columnName").toString();
                    }
                } else if (Objects.equals(dataSourceDTO.getType(), SQLSERVER.getCode())) {
                    if (Objects.equals(DataSourceConstant.SQLSERVER_KEY, map.get("columnKey"))) {
                        primaryKey = map.get("columnName").toString();
                    }
                }

                String columnName = map.get("columnName").toString();
                if (!"ID".equalsIgnoreCase(columnName) && (columnName.contains("id") || columnName.contains("ID"))) {
                    //带id(id除外)的字段去进行表名搜索
                    columnList.add(columnName);
                }
            }
            String foreignKey = null;
            if (StringUtils.isNotBlank(primaryKey)) {
                foreignKey = tableName + "_" + primaryKey;
                columnList.add(foreignKey);
            }
            if (!CollectionUtils.isEmpty(columnList)) {
                dataSourceDTO.setColumnNameList(columnList);
                ResultVO linkResultVO = databaseManager.getLinkTable(dataSourceDTO);
                taskService.setTaskProgress(taskId, 60.00);
                if (linkResultVO != null) {
                    List<Map<String, Object>> linkContent = linkResultVO.getContent();
                    List<String> tableNameList = new ArrayList<>();
                    linkContent.forEach(stringObjectMap -> {
                        if (!tableNameList.contains(stringObjectMap.get("tableName"))) {
                            tableNameList.add(stringObjectMap.get("tableName").toString());
                        }
                    });
                    for (String name : tableNameList) {
                        boolean isStrong = false;
                        if (Objects.equals(name, tableName)) {
                            //由于带id的字段会联想出中心表，所以这里做过滤
                            continue;
                        }
                        List<TableRelationDTO> tableRelationDTOList = new ArrayList<>();
                        ResultVO temp = new ResultVO();
                        temp.setTableName(name);
                        List<Map<String, Object>> subContent = new ArrayList<>();
                        for (Map<String, Object> map : linkContent) {
                            if (Objects.equals(name, map.get("tableName"))) {
                                subContent.add(map);
                                String columnName = map.get("columnName").toString();
                                if (columnList.contains(columnName)) {
                                    TableRelationDTO tableRelationDTO = new TableRelationDTO();
                                    tableRelationDTO.setCenterTable(tableName);
                                    tableRelationDTO.setLinkTable(name);
                                    tableRelationDTO.setLinkColumn(columnName);
                                    if (Objects.equals(columnName, foreignKey)) {
                                        isStrong = true;
                                        tableRelationDTO.setCenterColumn(primaryKey);
                                    } else {
                                        tableRelationDTO.setCenterColumn(columnName);
                                    }
                                    tableRelationDTOList.add(tableRelationDTO);
                                }
                            }

                        }
                        temp.setTableRelationList(tableRelationDTOList);
                        temp.setContent(subContent);
                        addStatisticsType(subContent);
                        //关联表排序逻辑实现
                        if (isStrong) {
                            strongLinkList.add(temp);
                        } else {
                            if (name.toLowerCase().startsWith(CommonConstant.DIM_TABLE_PREFIX)) {
                                dimLinkList.add(temp);
                            } else {
                                weakLinkList.add(temp);
                            }
                        }
                    }
                    if ((!CollectionUtils.isEmpty(strongLinkList)) && Objects.equals(Optional.ofNullable(dataSourceDTO.getIsDataWarehouse()).orElse(0), CommonConstant.NOT_WAREHOUSE)) {
                        list.addAll(strongLinkList);
                    }
                    if (!CollectionUtils.isEmpty(dimLinkList)) {
                        list.addAll(dimLinkList);
                    }
                    if ((!CollectionUtils.isEmpty(weakLinkList)) && Objects.equals(Optional.ofNullable(dataSourceDTO.getIsDataWarehouse()).orElse(0), CommonConstant.NOT_WAREHOUSE)) {
                        list.addAll(weakLinkList);
                    }
                }
            }
            TaskDTO taskDTO = taskService.get(taskId);
            taskDTO.setTaskId(taskId);
            taskDTO.setData(JSONObject.toJSONString(list));
            taskService.update(taskDTO);
            taskService.setTaskProgress(taskId, 100.00);
            return list;
        } catch (Exception e) {
            log.error("er图联想异步任务执行失败：{}", e.getMessage());
            TaskDTO taskDTO = taskService.get(taskId);
            taskDTO.setTaskId(taskId);
            taskDTO.setData("er图联想异步任务执行失败：" + e.getMessage());
            taskService.update(taskDTO);
            taskService.setTaskProgress(taskId, 100.00);
        }
        return null;
    }

    //checked
    @Override
    public DataSourceDTO get(Long dataSourceId, boolean joinTenant) {
        DataSource dataSource;
        if (joinTenant) {
            LambdaQueryWrapper<DataSource> query = Wrappers.<DataSource>lambdaQuery()
                    .eq(DataSource::getId, dataSourceId)
                    .and(wrapper -> wrapper
                            .eq(DataSource::getTenantId, loginUserManager.getCurrentTenantId())
                            .or()
                            .eq(DataSource::getTenantId, PermissionConstants.GLOBAL_BUSINESS_DATA_TENANT_ID_KEY));
            dataSource = InitiallyUtils.markInitially(() -> getOne(query));
        } else {
            dataSource = getById(dataSourceId);
        }
        if (Objects.isNull(dataSource)) {
            throw new DataFactoryException(ErrorCode.DATASOURCE_NOT_FOUND);
        }
        return toDataSourceDTO(dataSource);
    }

    @Override
    public DataSourceDTO getDataSource(Long dataSourceId) {
        DataSource dataSource = InitiallyUtils.withInitially(Wrappers.<DataSource>lambdaQuery().eq(DataSource::getId, dataSourceId),this::getOne, DataSource::getTenantId);
        if (Objects.isNull(dataSource)) {
            throw new DataFactoryException(ErrorCode.DATASOURCE_NOT_FOUND);
        }
        return toDataSourceDTONoPage(dataSource);
    }

    public DataSourceDTO getDataSourceMarkIgnore(Long dataSourceId) {
        QueryWrapper<DataSource> queryWrapper = new QueryWrapper();
        queryWrapper.eq("id", dataSourceId);
        return  toDataSourceDTONoPage(InitiallyUtils.markIgnore(() -> dataSourceMapper.selectOne(queryWrapper)));
    }

    private DataSourceDTO toDataSourceDTO(DataSource dataSource) {
        if (dataSource == null) {
            return null;
        }
        DataSourceDTO dataSourceDTO = UdmpBeanUtils.copy(dataSource, DataSourceDTO.class);
        dataSourceDTO.setDataSourceId(dataSource.getId());
        List<String> list = JSONObject.parseArray(dataSource.getDatabaseName(), String.class);
        dataSourceDTO.setDatabaseNameList(list);
        return dataSourceDTO;
    }

    private DataSourceDTO toDataSourceDTONoPage(DataSource dataSource) {
        if (dataSource == null) {
            return null;
        }
        DataSourceDTO dataSourceDTO = UdmpBeanUtils.copy(dataSource, DataSourceDTO.class);
        if (null == dataSourceDTO.getIsDataWarehouse() || CommonConstant.IS_WAREHOUSE != dataSourceDTO.getIsDataWarehouse()) {
            dataSourceDTO.setIsDataWarehouse(0);
        }
        //start##
        //适配isLocked的新结构
        dataSourceDTO.setDataSourceId(dataSource.getId());
        if (StringUtils.isEmpty(dataSource.getDatabaseConfig())) {
            if (!BeanUtil.isEmpty(dataSourceDTO.getDatabaseNameList())) {
                dataSource.setDatabaseConfig(JSON.toJSONString(dataSourceDTO.getDatabaseNameList()
                        .parallelStream().map(name -> new DatabasePair(name, 1)).collect(Collectors.toList())));
            }
        }
        dataSourceDTO.setDatabaseList(JSON.parseArray(dataSource.getDatabaseConfig(), DatabasePair.class));
        //end
        List<String> list;
        list = JSONObject.parseArray(dataSource.getDatabaseName(), String.class);
        dataSourceDTO.setDatabaseNameList(list);
        if (StringUtils.isEmpty(dataSource.getDatabaseConfig())) {
            List<DatabasePair> databasePairs = JSON.parseArray(dataSource.getDatabaseName(), String.class).parallelStream().map(name -> new DatabasePair(name, 1)).collect(Collectors.toList());
            dataSourceDTO.setDatabaseList(databasePairs);
        } else {
            dataSourceDTO.setDatabaseList(JSON.parseArray(dataSource.getDatabaseConfig(), DatabasePair.class));
        }
        return dataSourceDTO;
    }


    private void addStatisticsType(List<Map<String, Object>> content) {
        if (!CollectionUtils.isEmpty(content)) {
            content.forEach(stringObjectMap -> {
                Object dataType = stringObjectMap.get("dataType");
                if (dataType != null && AbstractDatabase.DATA_TYPE_TO_STATISTICS_TYPE.contains(dataType.toString().toUpperCase())) {
                    stringObjectMap.put("statisticsType", CommonConstant.STATISTICS_TYPE_MEASURE);
                    stringObjectMap.put("isNumber", CommonConstant.IS_NUMBER);
                } else {
                    stringObjectMap.put("statisticsType", CommonConstant.STATISTICS_TYPE_DIMENSION);
                    stringObjectMap.put("isNumber", CommonConstant.NOT_NUMBER);
                }
            });
        }
    }

    @Override
    public FormDataSourceDTO searchTables(Long dataSourceId, String table) {
//        List<String> databaseNames = getDatabaseList(dataSourceId);
//        DataSourceDTO dataSourceDTO = get(dataSourceId);
//        BaseSource source = SourceFactory.getSource(DataSourceTypeEnum.getType(dataSourceDTO.getType()).getName());
//        return source.searchTables(dataSourceDTO, table, databaseNames);
        return null;
    }

    @Override
    public JSONArray listDataType(Long dataSourceId) {
        DataSource dataSource = get(dataSourceId, true);

        DataSourceDTO dataSourceDTO = new DataSourceDTO();
        dataSourceDTO.setType(dataSource.getType());
        AbstractDatabase db = DatabaseManager.findDb(dataSourceDTO);

        Map<String, String> dataType = db.getColumnType();
//        BaseSource source = SourceFactory.getSource(DataSourceTypeEnum.getType(dataSource.getType()).getName());
//        Map<String, String> dataType = source.getType();
        Object[] strings = dataType.keySet().toArray();
        JSONArray jsonArray = new JSONArray();
        for (int i = 0; i < strings.length; i++) {
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("name", strings[i]);
            jsonObject.put("code", i);
            jsonArray.add(jsonObject);
        }
        return jsonArray;
    }

    @Override
    public void renameTable(TableVO tableVO) {
        DataSourceDTO dataSourceDTO = toDataSourceDTO(get(tableVO.getDataSourceId(), true));
        dataSourceDTO.setSelectDatabaseName(tableVO.getDatabaseName());
        databaseManager.renameTable(dataSourceDTO, tableVO);
//        BaseSource source = SourceFactory.getSource(DataSourceTypeEnum.getType(dataSourceDTO.getType()).getName());
//        source.renameTable(dataSourceDTO, tableVO);
    }


    @Override
    public List<Map<String,Object>> getTreeTable(long dataSourceId, String databaseName, boolean joinTenant) {
        ResultVO resultVO = getTableListWithAlias(dataSourceId, databaseName, joinTenant);
        return resultVO.getContent();
    }

    @Override
    public Integer getOutDataSourceCount() {
        LambdaQueryWrapper<DataSource> queryWrapper = new LambdaQueryWrapper<>();
//        queryWrapper.eq(DataSource::getIsDelete, CommonConstant.NOT_DEL);
//        queryWrapper.ne(DataSource::getIsDataWarehouse, CommonConstant.IS_WAREHOUSE);
        return InitiallyUtils.markInitially(()->dataSourceMapper.selectCount(queryWrapper));
    }

    private void checkIsLocked(DataSourceDTO dataSourceDTO, String databaseName) {
        if (CollectionUtils.isEmpty(dataSourceDTO.getDatabaseNameList())) {
            throw new DataFactoryException(ErrorCode.NULL_DATABASE);
        }
        if (StringUtils.isEmpty(dataSourceDTO.getDatabaseConfig())) {
            throw new DataFactoryException(ErrorCode.DATABASE_LOCKED);
        }
        List<DatabasePair> list = JSON.parseArray(dataSourceDTO.getDatabaseConfig(), DatabasePair.class);
        long count = list.parallelStream()
                .filter(pair -> Objects.equals(databaseName, pair.getDatabaseName())
                        && Objects.equals(CommonConstant.NOT_LOCKED, pair.getIsLocked()))
                .count();
        if (count <= 0) {
            throw new DataFactoryException(ErrorCode.DATABASE_LOCKED);
        }
    }

    /************* for OS 3.0 *****************/

    @Override
    public List<Map> importData(MultipartFile file, List<ExcelDataVo> params) {
        //检测文件合法性
        String filename = file.getOriginalFilename();
        if (!UDMPFileUtils.checkExcelFile(filename)) {
            throw new DataFactoryException(ErrorCode.FILE_TYPE_NOT_SUPPORT);
        }
        long fileSize = file.getSize();
        if (fileSize > CommonConstant.MAX_FILE_SIZE) {
            throw new DataFactoryException(ErrorCode.FILE_SIZE_EXCEED_LIMIT);
        }
        List<ExcelDataVo> data;
        try {
            //提取文件的信息
            data = UDMPFileUtils.parseExcel(file.getInputStream());
        } catch (Exception e) {
            throw new DataFactoryException(ErrorCode.EXCEL_PARSE_ERROR);
        }
        List<Map> finalRes = new ArrayList<>();
        for (int i = 0; i < params.size(); i++) {
            List<String> result = execImportData(params.get(i), data.get(i));
            finalRes.add(ImmutableMap.of(
                    "tableName", params.get(i).getTableName() == null ? "sheet1" : params.get(i).getTableName(),
                    "result", result));
        }
        return finalRes;
    }

    @Transactional(rollbackFor = Exception.class)
    protected List<String> execImportData(@NotNull ExcelDataVo param, ExcelDataVo data) {
        List<String> res;
        DataSourceDTO dataSource = Optional.ofNullable(getDataSource(param.getDataSourceId()))
                .orElseThrow(() -> new DataFactoryException(ErrorCode.DATASOURCE_NOT_FOUND));
        dataSource.setSelectDatabaseName(param.getDatabaseName());

        List<Pair<ColumnInfoVo, Integer>> mappings = new ArrayList<>();
        List<ColumnInfoVo> infos = param.getColumnInfoList();
        for (int i = 0; i < infos.size(); i++) {
            final ColumnInfoVo col = infos.get(i);
            if (col.isSelected()) {
                mappings.add(ImmutablePair.of(col, i));
            }
        }
        if (mappings.size() <= 0) {
            throw new DataFactoryException(ErrorCode.MUST_CHOOSE_FIELD);
        }

        switch (param.getMethod()) {
            case DataSourceConstant.IMPORT_METHOD_NEW:
                res = importNew(param, data, mappings, dataSource);
                break;
            case DataSourceConstant.IMPORT_METHOD_CLEAR:
                res = importClear(param, data, mappings, dataSource);
                break;
            case DataSourceConstant.IMPORT_METHOD_APPEND:
                res = importAppend(param, data, mappings, dataSource);
                break;
            default:
                throw new DataFactoryException(ErrorCode.PARAM_METHOD_ILLEGAL);
        }
        return res;
    }

    private List<String> importAppend(@NotNull ExcelDataVo param, ExcelDataVo data, List<Pair<ColumnInfoVo, Integer>> mappings, DataSourceDTO dataSource) {
        try {
            boolean isExisted = getTableList(param.getDataSourceId(), param.getDatabaseName(), true)
                    .getContent()
                    .stream()
                    .anyMatch(table -> table.get("tableName").equals(param.getDataTableName()));
            if (!isExisted) {
                throw new DataFactoryException(ErrorCode.TABLE_NOT_EXIST);
            }
            //插入数据
            return databaseManager.insert(dataSource, mappings, data.getColumnValueList(), param);
        } catch (DataFactoryException e) {
            log.error(e.getMessage());
            throw new DataFactoryException(ErrorCode.DATA_IMPORT_FAILED_APPEND);
        }
    }

    private List<String> importClear(@NotNull ExcelDataVo param, ExcelDataVo data, List<Pair<ColumnInfoVo, Integer>> mappings, DataSourceDTO dataSource) {
        AbstractDatabase db = DatabaseManager.findDb(dataSource);
        String tableName = db.escapeTbName(param.getDataTableName(), param.getDatabaseName());
        try {
            //删除表
            databaseManager.delete(dataSource, param);
            //新建表
            databaseManager.create(dataSource, mappings, param);
            //插入数据
            return databaseManager.insert(dataSource, mappings, data.getColumnValueList(), param);
        } catch (DataFactoryException e) {
            log.error(e.getMessage());
            try {
                //删除表
                databaseManager.delete(dataSource, param);
            } catch (Exception e1) {
                log.error("表格{}回滚失败,原因：{}", tableName, DataFactoryException.getMessage(e1));
            }
            throw new DataFactoryException(ErrorCode.DATA_IMPORT_FAILED_CLEAR);
        }
    }

    private List<String> importNew(@NotNull ExcelDataVo param, ExcelDataVo data, List<Pair<ColumnInfoVo, Integer>> mappings, DataSourceDTO dataSource) {
        AbstractDatabase db = DatabaseManager.findDb(dataSource);
        String tableName = db.escapeTbName(param.getTableName(), param.getDatabaseName());

        try {
            boolean isExisted = getTableList(param.getDataSourceId(), param.getDatabaseName(), true)
                    .getContent()
                    .stream()
                    .anyMatch(table -> table.get("tableName").equals(param.getDataTableName()));
            if (isExisted) {
                throw new DataFactoryException(ErrorCode.TABLE_ALREADY_EXIST);
            }
            //新建表
            databaseManager.create(dataSource, mappings, param);
            //插入数据
            return databaseManager.insert(dataSource, mappings, data.getColumnValueList(), param);
        } catch (DataFactoryException e) {
            log.error(e.getMessage());
            try {
                //删除表
                databaseManager.delete(dataSource, param);
            } catch (Exception e1) {
                log.error("表格{}回滚失败,原因：{}", tableName, DataFactoryException.getMessage(e1));
            }
            throw new DataFactoryException(ErrorCode.DATA_IMPORT_FAILED_NEW);
        }
    }


    @Override
    public List<String> schemas(Long dataSourceId, String databaseName) {
        DataSourceDTO dataSourceDTO = get(dataSourceId, false);
        dataSourceDTO.setSelectDatabaseName(databaseName);
        AbstractDatabase db = DatabaseManager.findDb(dataSourceDTO);
        if (Objects.equals(dataSourceDTO.getType(), SQLSERVER.getCode())) {
            return ((SqlServerDatabase) db).schemas(dataSourceDTO);
        } else if (Objects.equals(dataSourceDTO.getType(), PGSQL.getCode())) {
            return ((PGSQLDatabase) db).schemas(dataSourceDTO);
        } else if (Objects.equals(dataSourceDTO.getType(), KINGBASE.getCode())) {
            return ((KingbaseDatabase) db).schemas(dataSourceDTO);
        } else {
            return null;
        }
    }

    @Override
    public List<DataSource> exportDataSourceWithIds(Set<Long> ids) {
        if (ids == null || ids.isEmpty()) {
            return emptyList();
        }
        LambdaQueryWrapper<DataSource> dataSourceQW = new LambdaQueryWrapper<DataSource>();
        dataSourceQW.in(DataSource::getId, ids);
        return this.list(dataSourceQW);
    }

    @Override
    public void importDataSources(List<DataSource> dataSources) {
        if (dataSources == null || dataSources.isEmpty()) {
            return;
        }
        Set<Long> ids = dataSources.stream().map(DataSource::getId).collect(Collectors.toSet());
        LambdaQueryWrapper<DataSource> dataSourceQW = new LambdaQueryWrapper<DataSource>();
        dataSourceQW.in(DataSource::getId, ids);
        this.remove(dataSourceQW);
        this.saveBatch(dataSources);
    }

    @Override
    public IPage<DataSourceItem> pageDataSource(DataSource dataSource, PageRequest pageRequest) {

        final LambdaQueryWrapper<DataSource> query = Wrappers.lambdaQuery(dataSource)
                .eq(DataSource::getTenantId, loginUserManager.getCurrentTenantId())
                .or()
                .eq(DataSource::getTenantId, PermissionConstants.GLOBAL_BUSINESS_DATA_TENANT_ID_KEY);
        IPage<DataSource> page;
        try {
             page = InitiallyUtils.markInitially(() -> this.page(pageRequest.toPage(), query));
        } catch (Exception e) {
            throw new DataFactoryException(ErrorCode.SQL_EXEC_FAILED, "请检查排序字段规则是否正确");
        }

        List<DataSource> dataSources = page.getRecords();
        IPage<DataSourceItem> dataSourceItemIPage = new Page<>();
        BeanUtils.copyProperties(page, dataSourceItemIPage);
        List<DataSourceItem> records = new ArrayList<>();
        dataSources.stream().forEach(el -> {
            DataSourceItem item = new DataSourceItem();
            BeanUtils.copyProperties(el, item);
//            item.setIsAllSuccess(CommonConstant.TRUE);
            List<DatabaseItem> databaseList = new ArrayList<>();
            JSONArray.parseArray(el.getDatabaseName(), String.class).stream().forEach(database -> {
//                List<String> schemas = null;
                DatabaseItem databaseItem = new DatabaseItem();
//                try {
//                    schemas = schemas(el.getId(), database);
//                } catch (DataFactoryException e) {
//                    log.error("数据源{}获取模式出现错误：{}", el.getId() + "#" + database, e);
//                    databaseItem.setDatabaseName(database);
////                    databaseItem.setIsAllSuccess(CommonConstant.FALSE);
////                    item.setIsAllSuccess(CommonConstant.FALSE);
//                    databaseList.add(databaseItem);
//                    item.setDatabaseList(databaseList);
//                    return;
//                }
//                if (CollectionUtils.isEmpty(schemas)) {
                    databaseItem.setDatabaseName(database);
//                    List<String> tables;
//                    try {
//                        tables = getTreeTable(el.getId(), database, false);
//                    } catch (DataFactoryException e) {
//                        log.error("数据库{}获取数据表出现错误：{}", el.getId() + "#" + database, e);
//                        databaseItem.setIsAllSuccess(CommonConstant.FALSE);
//                        item.setIsAllSuccess(CommonConstant.FALSE);
//                        databaseList.add(databaseItem);
//                        return;
//                    }
//                    databaseItem.setTableList(tables);
//                    databaseItem.setIsAllSuccess(CommonConstant.TRUE);
                    databaseList.add(databaseItem);
//                } else {
//                    schemas.parallelStream().forEach(schema -> {
//                        String str = database + "." + schema;
////                        List<String> tables;
////                        try {
////                            tables = getTreeTable(el.getId(), str, false);
////                        } catch (DataFactoryException e) {
////                            log.error("数据库{}获取数据表出现错误：{}", el.getId() + "#" + str, e);
////                            databaseItem.setIsAllSuccess(CommonConstant.FALSE);
////                            item.setIsAllSuccess(CommonConstant.FALSE);
////                            databaseList.add(databaseItem);
////                            return;
////                        }
//                        databaseItem.setDatabaseName(str);
////                        databaseItem.setTableList(tables);
//                        databaseList.add(databaseItem);
//                    });
//                }
            });
            item.setDatabaseList(databaseList);
            records.add(item);
        });
        dataSourceItemIPage.setRecords(records);
        return dataSourceItemIPage;
    }
}
